package com.mjf.orderpay_detect

import java.net.URL

import com.mjf.dim.{OrderEvent, OrderResult}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * ProcessFunction 实现订单超时检测
 */
object OrderTimeoutNotCEP {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val resource: URL = getClass.getResource("/OrderLog.csv")
    val inputStream: DataStream[String] = env.readTextFile(resource.getPath)

    val orderEventStream: DataStream[OrderEvent] = inputStream
      .map {
        line =>
          val dataArr: Array[String] = line.split(",")
          OrderEvent(dataArr(0).toLong, dataArr(1), dataArr(2), dataArr(3).toLong)
      }
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(3)) {
        override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
      })

    // 自定义ProcessFunction,做精细化的流程控制
    val orderResultStream: DataStream[OrderResult] = orderEventStream
      .keyBy(_.orderId)
      .process(new OrderPayDetect())

    // 打印输出
    orderResultStream.print()
    orderResultStream.getSideOutput(new OutputTag[OrderResult]("timeout")).print("timeout")

    env.execute("OrderTimeoutNotCEP")

  }
}

// 主流输出正常支付的订单，侧输出流输出超时报警订单
class OrderPayDetect() extends KeyedProcessFunction[Long, OrderEvent, OrderResult] {

  // 定义状态，用来保存是否来过create和pay事件的标识位，以及定时器时间戳
  lazy val isCreatedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("is-created", classOf[Boolean]))
  lazy val isPayedState: ValueState[Boolean] = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("is-payed", classOf[Boolean]))
  lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timer-ts", classOf[Long]))

  val orderTimeoutOutputTag: OutputTag[OrderResult] = new OutputTag[OrderResult]("timeout")

  override def processElement(value: OrderEvent, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#Context, out: Collector[OrderResult]): Unit = {

    // 先取出当前状态
    val isCreated: Boolean = isCreatedState.value()
    val isPayed: Boolean = isPayedState.value()
    val timerTs: Long = timerTsState.value()

    // 判断当前事件的类型，分成不同的情况讨论
    // 情况1：来的是create,要继续判断之前是否有pay来过
    if(value.eventType == "create") {
      // 情况1.1：如果已经pay过的话，匹配成功，输出到主流，清空状态
      if(isPayed) {
        out.collect(OrderResult(value.orderId, "payed successful"))
        isPayedState.clear()
        timerTsState.clear()
        ctx.timerService().deleteEventTimeTimer(timerTs)
      } else {
        // 情况1.2：如果没有pay过，那么就注册一个15分钟后的定时器，开始等待
        val ts: Long = value.eventTime * 1000L + 15 * 60 * 1000L
        ctx.timerService().registerEventTimeTimer(ts)
        timerTsState.update(ts)
        isCreatedState.update(true)
      }
    } else if(value.eventType == "pay") { // 情况2：来的是pay,要继续判断之前是否有create来过
      // 情况2.1：如果create已经来过，匹配成功，要继续判断间隔时间是否超过了15分钟
      if(isCreated) {
        // 情况2.1.1：如果没有超时，正常输出到主流
        if(value.eventTime * 1000L < timerTs) {
          out.collect(OrderResult(value.orderId, "payed successful"))
        } else {
          // 情况2.1.2：如果已经超时，输出timeout报警到侧输出流
          ctx.output(orderTimeoutOutputTag, OrderResult(value.orderId, "payed but already timeout"))
        }
        // 不论哪种情况，都已经有了输出，清空状态
        isCreatedState.clear()
        timerTsState.clear()
        ctx.timerService().deleteEventTimeTimer(timerTs)
      } else {
        // 情况2.2：如果create没来，需要等待乱序create,注册一个当前pay时间抽的定时器
        val ts: Long = value.eventTime * 1000L
        ctx.timerService().registerEventTimeTimer(ts)
        timerTsState.update(ts)
        isPayedState.update(true)
      }
    }

  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, OrderEvent, OrderResult]#OnTimerContext, out: Collector[OrderResult]): Unit = {

    // 定时器触发，要判断是哪种情况
    if(isPayedState.value()) {
      // 如果pay过，那么说明create没来，可能出现数据丢失异常情况
      ctx.output(orderTimeoutOutputTag, OrderResult(ctx.getCurrentKey, "already payed but not found create"))
    } else {
      // 如果没有 pay 过，那么说明真正15分钟超时
      ctx.output(orderTimeoutOutputTag, OrderResult(ctx.getCurrentKey, "order timeout"))
    }

    // 清理状态
    isPayedState.clear()
    isCreatedState.clear()
    timerTsState.clear()

  }
}
